package com.passion.mq.rocketmq.queue1;

import com.passion.mq.config.RocketMqConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Service;

/**
 * 类描述：rocketmq的消息接收器(简单方式)
 * 这种写法是不使用事务的，即消费消息自动确认的，对于需要考虑事务的场景不适用
 * 它是单条消费的
 * @author ZhuYin
 * @since 2024年12月30日
 */
@Slf4j
@Service
@ConditionalOnBean(RocketMqConfig.class)
@RocketMQMessageListener(
        topic = "${archives.mq.business[0].queueName}",
        consumerGroup = "${rocketmq.consumer.group}",
        consumeMode = ConsumeMode.ORDERLY, // 或者 ConsumeMode.CONCURRENTLY
        // 消息消费最大尝试次数
        maxReconsumeTimes = 3,
        messageModel = MessageModel.CLUSTERING // 或者 MessageModel.BROADCASTING
)
public class UserOperationQueueSingleMsgConsumer implements RocketMQListener<MessageExt> {

    @Override
    public void onMessage(MessageExt message) {
        try {
            String messageBody = new String(message.getBody());
            System.out.println("Received message: " + messageBody);

            // 执行业务逻辑
            processBusinessLogic(messageBody);

            // 如果业务逻辑成功，则无需做额外操作，消息会被自动确认
        } catch (Exception e) {
            // 如果业务逻辑失败，抛出异常以触发重试机制
            log.error("Failed to process message: {}", message.getMsgId(), e);
            throw new RuntimeException("Business logic failed");
        }
    }

    private void processBusinessLogic(String messageBody) throws Exception {
        // 你的业务逻辑处理代码
        if ("success".equals(messageBody)) {
            // 模拟成功处理
        } else {
            // 模拟失败处理
            throw new RuntimeException("Business logic failed");
        }
    }

}
